package io.gitee.hechenl.datamultitrans.mqtt;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class MqttClientInitiator {

    @AllArgsConstructor
    @Data
    public static class ClientWrapper {
        private InitClient initClient;
        private MqttClient mqttClient;
    }

    private final Map<String, ClientWrapper> clientMap = new ConcurrentHashMap<>();

    public void init(InitClient ...list) throws Exception {
        List<InitClient> clients = Arrays.asList(list);
        if (containsInValid(clients)) {
            throw new IllegalArgumentException("init failed, clients invalid");
        }
        for (InitClient client : clients) {
            clientMap.put(client.getClientId(), new ClientWrapper(client, initMqttClient(client)));
        }
    }

    private boolean containsInValid(List<InitClient> clients) {
        if (CollectionUtils.isEmpty(clients)) {
            return false;
        }
        return clients.stream().anyMatch(
                v -> (!StringUtils.hasText(v.getClientId()))
                        || (!StringUtils.hasText(v.getBrokeAddress())
                                || !StringUtils.hasText(v.getBrokeUserName())
                                || !StringUtils.hasText(v.getBrokePassword())));
    }

    private MqttClient initMqttClient(InitClient initClient) throws Exception {
        MqttClient mqttClient = MqttUtil.createClient(initClient.getClientId(), initClient.getBrokeAddress());
        mqttClient.connect(MqttUtil.createConnectOptions(initClient.getCleanSession(), initClient.getBrokeUserName(), initClient.getBrokePassword()));
        mqttClient.setCallback(initClient.getCallback());
        if (StringUtils.hasText(initClient.getSubTopic())) {
            mqttClient.subscribe(initClient.getSubTopic());
        }
        return mqttClient;
    }

    @PreDestroy
    public void shutdown() {
        if (!CollectionUtils.isEmpty(clientMap)) {
            clientMap.forEach((clientId, client) -> {
                try {
                    MqttClient mqttClient = client.getMqttClient();
                    mqttClient.disconnect();
                    mqttClient.close();
                } catch (Exception e) {
                    log.error("Shutdown mqtt client disconnect error, clientId:{}", clientId, e);
                }
            });
        }
    }

    public void publish(Integer qos, String pubTopic, String content) {
        try {
            Optional<ClientWrapper> wrapper = clientMap.values().stream().filter(
                    v -> Objects.equals(v.getInitClient().getPubTopic(), pubTopic)).findFirst();
            if (!wrapper.isPresent()) {
                throw new RuntimeException("Mqtt publish failed, Not client for pubTopic:" + pubTopic);
            }
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(qos);
            wrapper.get().getMqttClient().publish(pubTopic, message);
        } catch (Exception e) {
            log.error("Mqtt publish error, qos:{}, pubTopic:{}, content:{}", qos, pubTopic, content, e);
        }
    }
}